package org.niit.kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/*
    使用异步的方式生产数据
            对数据发送到Kafka服务要进行一个回复

 */
public class KafkaProducerTest2 {

    public static void main(String[] args) throws InterruptedException {

        //1.连接的配置信息
        Properties pros = new Properties();
        pros.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092");
        pros.put(ProducerConfig.ACKS_CONFIG,"all");
        pros.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        pros.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());

        //2.创建生产者
        KafkaProducer<String, String> producer = new KafkaProducer<>(pros);

        //3.进行异步生产数据  对生产数据后有结果反馈
        for (int i=0;i<100000;i++){
            //record 一条数据
            ProducerRecord<String, String> record = new ProducerRecord<>("BD1_1", null, i + "");

            producer.send(record, new Callback() {
                //Callback 回调函数 ，对Kafka服务器对生产结果进行应答，应答的内容会在onCompletion方法进行处理
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    /*
                            recordMetadata:消息的元数据，包含了主题 key 分区 偏移量
                              e：错误信息
                     */
                    //1.先判断是否发送成功
                    if(e == null){
                        //没有异常 表示发送成功了
                        String topic = recordMetadata.topic();
                        int partition = recordMetadata.partition();
                        long timestamp = recordMetadata.timestamp();
                        long offset = recordMetadata.offset();
                        System.out.println("发送数据成功，该数据的主题："+topic+",分区："+partition
                        +",时间戳："+timestamp+",偏移量:"+offset);
                    }else {
                        //有异常发送失败
                        System.out.println("发送数据失败:"+e.getMessage());
                    }
                }
            });
            Thread.sleep(3000);
        }

        producer.close();



    }
}
